package com.shujia.flink.source

import java.sql.{Connection, Driver, DriverManager}

import Demo3ConnectorMysql.Student
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object Demo3ConnectorMysql {
  def main(args: Array[String]): Unit = {

    /**
      * 自定义source读取mysql数据
      *
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment


    //增加数据源
    val mysqlDS = env.addSource(new MysqlSource)

    mysqlDS.map(_.String).print()


    env.execute("Demo3ConnectorMysql")

  }

  case class Student(String: String, name: String, age: Int, gender: String, clazz: String)

}

/**
  * RichSourceFunction  单一的source
  * RichParallelSourceFunction   并行的source
  *
  * 两种source需要实现的方法都是一样的
  *
  */

class MysqlSource extends RichSourceFunction[Student] {


  var connector: Connection = _

  //在run 之前执行
  override def open(parameters: Configuration): Unit = {
    //打开连接
    Class.forName("com.mysql.jdbc.Driver")
    connector = DriverManager.getConnection("jdbc:mysql://node1:3306/student", "root", "123456")

  }


  //在run犯法之后执行
  override def close(): Unit = {
    if (connector != null) {
      connector.close()
    }
  }

  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {


    val sql = "select * from student limit 10"

    val stat = connector.createStatement()

    val resultSet = stat.executeQuery(sql)

    while (resultSet.next()) {
      val id = resultSet.getString(1)
      val name = resultSet.getString(2)
      val age = resultSet.getInt(3)
      val gedner = resultSet.getString(4)
      val clazz = resultSet.getString(5)

      val stu = Student(id, name, age, gedner, clazz)

      //发送到下游
      ctx.collect(stu)
      Thread.sleep(1000)
    }
  }

  //在关闭程序的时候执行
  override def cancel(): Unit = {
  }
}
